﻿using FastEmit;
using Flurl.Http;
using OkFlurl;
using System.IO.Compression;
using System.Runtime.CompilerServices;
using System.Text;
using System.Text.Encodings.Web;
using System.Text.Json;
using System.Text.Json.Nodes;
using System.Text.Json.Serialization;

namespace VictoriaMetrics.Client
{
    //https://docs.victoriametrics.com
    //https://docs.victoriametrics.com/metricsql
    //https://docs.victoriametrics.com/url-examples
    //https://docs.victoriametrics.com/#json-line-format
    //https://docs.victoriametrics.com/#how-to-export-data-in-json-line-format
    //https://docs.victoriametrics.com/single-server-victoriametrics/#prometheus-querying-api-usage
    //https://docs.victoriametrics.com/keyconcepts/#query-data
    //https://docs.victoriametrics.com/metricsql
    //-search.latencyOffset=30  导致插入数据查询延时
    public class VictoriaMetricsClient
    {
        #region static

        public static readonly byte[] N = [(byte)'\n'];

        //毫秒序列化
        public static readonly JsonSerializerOptions JsonOptionsMs = new()
        {
            Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping,
            NumberHandling = JsonNumberHandling.AllowReadingFromString,
            PropertyNameCaseInsensitive = true,
            PropertyNamingPolicy = null,
            DictionaryKeyPolicy = null
        };

        static VictoriaMetricsClient()
        {
            JsonOptionsMs.Converters.Add(new DateTimeMsConverter());
            JsonOptionsMs.Converters.Add(new DateTimeOffsetMsConverter());
        }

        #endregion

        #region MyRegion

        public FlurlRequest CreateRequest(string path)
        {
            var req = Api.CreateRequest(path);

            if (AccountID != 0)
            {
                req.WithHeader("AccountID", AccountID);
            }

            if (ProjectID != 0)
            {
                req.WithHeader("ProjectID", ProjectID);
            }

            return req;
        }

        #endregion

        public int AccountID { get; set; }

        public int ProjectID { get; set; }

        public FlurlApi Api { get; }

        //VictoriaMetrics Single: 8428   vmagent: 8429  vmauth: 8427  vminsert: 8480
        public VictoriaMetricsClient(string url, string username = null, string password = null, int? timeout = null)
        {
            Api = new FlurlApi(url)
            {
                UserName = username,
                Password = password,
                Timeout = timeout,
                JsonOptionsDefault = JsonOptionsMs
            };
        }

        public Task ForceFlushAsync(CancellationToken token = default)
        {
            var req = CreateRequest("/internal/force_flush");
            return Api.GetStringAsync(req, null, token);
        }

        //YYYY_MM  2020_08
        public Task ForceMergeAsync(string partition_prefix, CancellationToken token = default)
        {
            var req = CreateRequest("/internal/force_merge");
            req.SetQueryParam("partition_prefix", partition_prefix);
            return Api.GetStringAsync(req, null, token);
        }

        public Task DeleteAsync(string match, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v1/admin/tsdb/delete_series");
            req.SetQueryParam("match[]=", match);
            return Api.GetStringAsync(req, null, token);
        }

        public async Task ImportAsync(Func<Stream, Task> func, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v1/import");
            req.WithHeader("Content-Type", "application/json");

            if (gzip)
            {
                req.WithHeader("Content-Encoding", "gzip");
            }

            using var fcs = new FlurlStreamContent(async stream =>
            {
                if (!gzip)
                {
                    await func(stream);
                    return;
                }

                using var gz = new GZipStream(stream, compressLv, true);
                await func(gz);
            });

            (await Api.PostAsync(req, fcs, token: token))?.Dispose();
        }

        public Task ImportJsonlineAsync(IEnumerable<string> list, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, int sendBuffer = 100, CancellationToken token = default)
        {
            return ImportAsync(async stream =>
            {
                var i = 0;
                var currentRead = 0;
                byte[] bytes;

                foreach (var item in list)
                {
                    i++;
                    currentRead++;
                    if (i > 1)
                    {
#if NET462
                        await stream.WriteAsync(N, 0, 1, token);
#else
                        await stream.WriteAsync(N, token);
#endif
                    }

                    bytes = Encoding.UTF8.GetBytes(item);
#if NET462
                    await stream.WriteAsync(bytes, 0, bytes.Length, token);
#else
                    await stream.WriteAsync(bytes, token);
#endif

                    if (currentRead == sendBuffer)
                    {
                        currentRead = 0;
                        await stream.FlushAsync(token);
                    }
                }

            }, gzip, compressLv, token);

        }

        public Task ImportMetricsAsync(IEnumerable<MetricsEntity> list, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, int sendBuffer = 100, CancellationToken token = default)
        {
            return ImportAsync(async stream =>
            {
                var i = 0;
                var currentRead = 0;
                foreach (var item in list)
                {
                    i++;
                    currentRead++;
                    if (i > 1)
                    {
#if NET462
                        await stream.WriteAsync(N, 0, 1, token);
#else
                        await stream.WriteAsync(N, token);
#endif
                    }

                    await JsonSerializer.SerializeAsync(stream, item, JsonOptionsMs, token);

                    if (currentRead == sendBuffer)
                    {
                        currentRead = 0;
                        await stream.FlushAsync(token);
                    }
                }

            }, gzip, compressLv, token);
        }

        public Task ImportPointAsync(IEnumerable<MetricsPointData> list, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, int sendBuffer = 100, CancellationToken token = default)
        {
            return ImportAsync(async stream =>
            {
                var i = 0;
                var currentRead = 0;
                foreach (var point in list)
                {
                    i++;
                    currentRead++;

                    if (i > 1)
                    {
#if NET462
                        await stream.WriteAsync(N, 0, 1, token);
#else
                        await stream.WriteAsync(N, token);
#endif
                    }

                    if (string.IsNullOrEmpty(point.MetricName))
                    {
                        throw new Exception("MetricName is null or empty.");
                    }

                    if (point.Metric == null || point.Metric.Count == 0)
                    {
                        throw new Exception("Metric is null or empty.");
                    }

                    if (!point.Metric.ContainsKey("__name__"))
                    {
                        point.Metric.Add("__name__", point.MetricName);
                    }

                    if (point.Time == null)
                    {
                        point.Time = DateTimeOffset.Now;
                    }

                    var par = new
                    {
                        metric = point.Metric,
                        values = new double[] { point.Value },
                        timestamps = new long[] { point.Time.Value.ToUnixTimeMilliseconds() },
                    };

                    await JsonSerializer.SerializeAsync(stream, par, JsonOptionsMs, token);

                    if (currentRead == sendBuffer)
                    {
                        currentRead = 0;
                        await stream.FlushAsync(token);
                    }
                }

            }, gzip, compressLv, token);
        }

        public Task ImportEntityAsync<T>(string metricName, IEnumerable<T> list, bool gzip = false, CompressionLevel compressLv = CompressionLevel.Optimal, int sendBuffer = 100, CancellationToken token = default) where T : class
        {
            var t = typeof(T);
            if (t == typeof(object))
            {
                t = list.First().GetType();
            }
            var accssor = FastEmitFactory.GetTypeAccessor(t);
            var ch = CacheHelper.GetCache(t);

            return ImportAsync(async stream =>
            {
                var i = 0;
                var currentRead = 0;
                foreach (var item in list)
                {
                    foreach (var field in ch.FieldKeys)
                    {
                        i++;
                        currentRead++;
                        if (i > 1)
                        {
#if NET462
                            await stream.WriteAsync(N, 0, 1, token);
#else
                            await stream.WriteAsync(N, token);
#endif
                        }

                        var entity = new MetricsEntity(metricName);
                        foreach (var tag in ch.TagKeys)
                        {
                            var tagVal = (string)accssor.GetValue(item, tag);
                            if (string.IsNullOrEmpty(tagVal))
                            {
                                throw new Exception($"{t.FullName}==>{tag}==>is null or empty.");
                            }
                            entity.metric.Add(tag, tagVal);
                        }

                        var fieldVal = accssor.GetValue(item, field);
                        entity.metric.Add("_field", field);
                        entity.values.Add(Convert.ToDouble(fieldVal));

                        var timeVal = accssor.GetValue(item, ch.TimeKey);
                        if (timeVal == null)
                        {
                            entity.timestamps.Add(DateTimeOffset.Now.ToUnixTimeMilliseconds());
                        }
                        else
                        {
                            entity.timestamps.Add(((DateTimeOffset)timeVal).ToUnixTimeMilliseconds());
                        }

                        await JsonSerializer.SerializeAsync(stream, entity, JsonOptionsMs, token);

                        if (currentRead == sendBuffer)
                        {
                            currentRead = 0;
                            await stream.FlushAsync(token);
                        }
                    }
                }

            }, gzip, compressLv, token);
        }

        public async Task<Stream> ExportStreamAsync(string match, DateTimeOffset start, DateTimeOffset end, bool gzip = false, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v1/export");
            if (string.IsNullOrEmpty(match))
            {
                match = "{__name__!=\"\"}";
            }
            req.SetQueryParam("match[]", match);
            req.SetQueryParam("start", start.ToUnixTimeMilliseconds());
            req.SetQueryParam("end", end.ToUnixTimeMilliseconds());

            if (gzip)
            {
                req.WithHeader("Accept-Encoding", "gzip");
            }

            return await Api.GetStreamAsync(req, null, token);
        }

        //match={__name__=~"measurement_.*"}
        //match[]={__name__!=""}
        public async IAsyncEnumerable<string> ExportJsonline(string match, DateTimeOffset start, DateTimeOffset end, bool gzip = false, [EnumeratorCancellation] CancellationToken token = default)
        {
            var req = CreateRequest("/api/v1/export");
            if (string.IsNullOrEmpty(match))
            {
                match = "{__name__!=\"\"}";
            }
            req.SetQueryParam("match[]", match);
            req.SetQueryParam("start", start.ToUnixTimeMilliseconds());
            req.SetQueryParam("end", end.ToUnixTimeMilliseconds());

            if (gzip)
            {
                req.WithHeader("Accept-Encoding", "gzip");
            }

            using var stream = await Api.GetStreamAsync(req, null, token);
            using var sr = new StreamReader(stream);
            string? line;

#if NET462 || NET6_0
            while ((line = await sr.ReadLineAsync()) != null)
            {
                token.ThrowIfCancellationRequested();
                yield return line;
            }
#else
            while ((line = await sr.ReadLineAsync(token)) != null)
            {
                yield return line;
            }
#endif
        }

        public async IAsyncEnumerable<MetricsEntity> ExportMetrics(string match, DateTimeOffset start, DateTimeOffset end, bool gzip = false, [EnumeratorCancellation] CancellationToken token = default)
        {
            var unbuffer = ExportJsonline(match, start, end, gzip, token);
            await foreach (var line in unbuffer)
            {
                var mc = JsonSerializer.Deserialize<MetricsEntity>(line);
                yield return mc;
            }
        }

        public async IAsyncEnumerable<MetricsPointData> ExportPoint(string match, DateTimeOffset start, DateTimeOffset end, bool gzip = false, [EnumeratorCancellation] CancellationToken token = default)
        {
            var unbuffer = ExportMetrics(match, start, end, gzip, token);
            int i;
            await foreach (var mc in unbuffer)
            {
                i = 0;
                foreach (var val in mc.values)
                {
                    var point = new MetricsPointData
                    {
                        MetricName = mc.metric["__name__"],
                        Metric = mc.metric,
                        Value = val
                    };
                    var time = mc.timestamps[i];
                    point.Time = DateTimeOffset.FromUnixTimeMilliseconds(time).ToLocalTime();
                    i++;
                    yield return point;
                }
            }
        }

        public async Task<IEnumerable<T>> ExportListAsync<T>(string match, DateTimeOffset start, DateTimeOffset end, bool gzip = false, CancellationToken token = default) where T : class, new()
        {
            var t = typeof(T);
            var accssor = FastEmitFactory.GetTypeAccessor(t);
            var ch = CacheHelper.GetCache(t);

            Dictionary<string, T> dict = [];

            var unbuffer = ExportMetrics(match, start, end, gzip, token);
            int i;
            await foreach (var mc in unbuffer)
            {
                var hasField = mc.metric.TryGetValue("_field", out var field);
                if (hasField)
                {
                    hasField = ch.FieldKeys.Contains(field);
                }

#if NET462
                var id = string.Join("-", ch.TagKeys);
#else
                var id = string.Join('-', ch.TagKeys);
#endif

                i = 0;
                foreach (var val in mc.values)
                {
                    var time = mc.timestamps[i];
                    var timeOffset = DateTimeOffset.FromUnixTimeMilliseconds(time).ToLocalTime();

                    var uniId = $"{id}-{time}";

                    var ok = dict.TryGetValue(uniId, out var model);
                    if (!ok)
                    {
                        model = new T();
                        foreach (var tagkey in ch.TagKeys)
                        {
                            var exist = mc.metric.TryGetValue(tagkey, out var tagVal);
                            if (exist)
                            {
                                accssor.SetValue(model, tagkey, tagVal);
                            }
                        }
                        dict.Add(uniId, model);
                    }

                    accssor.SetValue(model, ch.TimeKey, timeOffset);

                    if (hasField)
                    {
                        accssor.SetValueConvert(model, field, val);
                    }

                    i++;
                }
            }

            return dict.Select(s => s.Value);
        }

        public Task<Stream> QueryStreamAsync(string query, DateTimeOffset start, DateTimeOffset end, string step = "5s", int timeout = 0, int latency_offset = 0, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v1/query_range");
            req.SetQueryParam("query", query);
            req.SetQueryParam("start", start.ToUnixTimeMilliseconds());
            req.SetQueryParam("end", end.ToUnixTimeMilliseconds());

            req.SetQueryParam("step", step);
            if (timeout > 0)
            {
                req.SetQueryParam("timeout", $"{timeout}s");
            }

            if (latency_offset > 0)
            {
                req.SetQueryParam("latency_offset", $"{latency_offset}s");
            }

            return Api.GetStreamAsync(req, null, token);
        }

        public Task<T> QueryAsync<T>(string query, DateTimeOffset start, DateTimeOffset end, string step = "5s", int timeout = 0, int latency_offset = 0, CancellationToken token = default)
        {
            var req = CreateRequest("/api/v1/query_range");
            req.SetQueryParam("query", query);
            req.SetQueryParam("start", start.ToUnixTimeMilliseconds());
            req.SetQueryParam("end", end.ToUnixTimeMilliseconds());

            req.SetQueryParam("step", step);
            if (timeout > 0)
            {
                req.SetQueryParam("timeout", $"{timeout}s");
            }

            if (latency_offset > 0)
            {
                req.SetQueryParam("latency_offset", $"{latency_offset}s");
            }

            return Api.GetJsonAsync<T>(req, null, token);
        }

        public async Task<List<QueryResult>> QueryResultAsync(string query, DateTimeOffset start, DateTimeOffset end, string step = "5s", int timeout = 0, int latency_offset = 0, CancellationToken token = default)
        {
            var obj = await QueryAsync<JsonObject>(query, start, end, step, timeout, latency_offset, token);
            var result = obj["data"]["result"].AsArray();
            List<QueryResult> list;
            if (result.Count > 0)
            {
                list = new List<QueryResult>(result.Count);
                foreach (JsonObject item in result)
                {
                    var model = new QueryResult();
                    model.metric = item["metric"].Deserialize<Dictionary<string, string>>();
                    var values = item["values"].AsArray();
                    if (values.Count > 0)
                    {
                        model.Values = new List<QueryValue>(values.Count);
                        foreach (JsonArray arr in values)
                        {
                            var v = new QueryValue
                            {
                                Time = DateTimeOffset.FromUnixTimeSeconds(arr[0].GetValue<long>()).ToLocalTime(),
                                Value = Convert.ToDouble(arr[1].GetValue<string>())
                            };
                            model.Values.Add(v);
                        }
                    }
                    else
                    {
                        model.Values = [];
                    }
                    list.Add(model);
                }
            }
            else
            {
                list = [];
            }

            return list;
        }
    }
}
